package cn.mengfly.testnrz.crazyIm.nettyserver;

import cn.mengfly.testnrz.crazyIm.ServerConstants;
import cn.mengfly.testnrz.curator.ZkClient;
import cn.mengfly.testnrz.json.JsonUtil;
import cn.mengfly.testnrz.util.ZookeeperUtil;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * IM 路由器
 *
 * @author Mengfly
 */
public class WorkerRouter {

    private CuratorFramework client = null;
    private static WorkerRouter singleInstance;

    private static final String PATH = ServerConstants.MANAGE_PATH;

    /**
     * 用于封装和保存所有的在线节点
     */
    private final ConcurrentMap<Long, WorkerReSender> workerMap = new ConcurrentHashMap<>();

    public static WorkerRouter inst() {
        if (null == singleInstance) {
            synchronized (WorkerRouter.class) {
                if (null == singleInstance) {
                    singleInstance = new WorkerRouter();
                    singleInstance.client = ZkClient.inst().getClient();
                    singleInstance.init();
                }
            }
        }
        return singleInstance;
    }

    private void init() {
        try {
            TreeCache treeCache = new TreeCache(client, PATH);
            TreeCacheListener l = (curatorFramework, event) -> {
                ChildData data = event.getData();
                if (data != null) {
                    switch (event.getType()) {
                        case NODE_REMOVED:
                            processNodeRemoved(data);
                            break;
                        case NODE_ADDED:
                            processNodeAdded(data);
                            break;
                        default:
                            break;
                    }
                } else {
                    System.out.println("[TreeCache] 节点数据为空, path=" + PATH);
                }
            };
            treeCache.getListenable().addListener(l);
            treeCache.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 节点移除处理
     *
     * @param data 节点
     */
    private void processNodeRemoved(ChildData data) {
        System.out.println("[TreeCache] 节点更新端口," +
                " path=" + data.getPath() +
                ",data=" + new String(data.getData(), StandardCharsets.UTF_8));
        // 获取路由
        String path = data.getPath();
        long id = ZookeeperUtil.getId(path, ServerConstants.PATH_PREFIX);
        if (workerMap.containsKey(id)) {
            final WorkerReSender workerReSender = workerMap.get(id);
            // 断开连接
            workerReSender.disConnect();
        }
        workerMap.remove(id);
    }

    /**
     * 节点增加的处理
     * <p>
     * 表示有了新的Netty服务器加入了集群
     *
     * @param data data
     */
    private void processNodeAdded(ChildData data) {
        System.out.println("[TreeCache] 节点更新端口," +
                " path=" + data.getPath() +
                ",data=" + new String(data.getData(), StandardCharsets.UTF_8));
        byte[] payload = data.getData();
        String path = data.getPath();
        ImNode imNode = JsonUtil.jsonBytes2Object(payload, ImNode.class);
        long id = ZookeeperUtil.getId(path, ServerConstants.PATH_PREFIX);
        imNode.setId(id);

        WorkerReSender reSender = workerMap.get(imNode.getId());
        // 判断服务器是否已经被添加过了
        if (null != reSender && reSender.getRemoteNode().equals(imNode)) {
            return;
        }
        // 创建一个消息转发器
        reSender = new WorkerReSender(imNode);
        reSender.doConnect();
        workerMap.put(id, reSender);
    }
}
